"""
BaoStock日K数据服务层

本文件提供使用BaoStock获取股票日K数据的服务方法。
相关知识点：
- BaoStock数据获取
- 异步编程
- 数据格式转换
"""
from datetime import date, datetime, timedelta
from typing import List, Optional, Set, Tuple
from decimal import Decimal
import baostock as bs
import pandas as pd
from fastapi.concurrency import run_in_threadpool
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from exceptions.exception import ServiceException
from module_admin.entity.vo.common_vo import CrudResponseModel
from module_stock.dao.baostock_daily_dao import BaoStockDailyDao
from module_stock.entity.do.baostock_daily_do import StockBaoStockDaily
from module_stock.entity.vo.baostock_daily_vo import BaoStockDailyModel
from utils.common_util import CamelCaseUtil
from utils.log_util import logger


class BaoStockDailyService:
    """
    BaoStock股票日线模块服务层
    """

    @staticmethod
    def _convert_symbol_to_baostock_code(symbol: str) -> str:
        """
        将股票代码转换为BaoStock格式
        
        BaoStock代码格式：
        - 上海: sh.600000
        - 深圳: sz.000001
        
        :param symbol: 股票代码，如 "600000" 或 "000001"
        :return: BaoStock格式代码，如 "sh.600000" 或 "sz.000001"
        """
        if not symbol:
            raise ValueError("股票代码不能为空")
        
        symbol = symbol.strip()
        
        # 判断是上海还是深圳
        if symbol.startswith(('600', '601', '603', '605', '688', '689')):
            return f"sh.{symbol}"
        elif symbol.startswith(('000', '001', '002', '003', '300')):
            return f"sz.{symbol}"
        else:
            raise ValueError(f"不支持的股票代码格式: {symbol}")

    @staticmethod
    def _safe_float(x, default=None):
        """
        安全地将值转换为float
        
        :param x: 待转换的值
        :param default: 转换失败时的默认值
        :return: float值或default
        """
        try:
            if x is None or x == '' or x == 'nan':
                return default
            return float(x)
        except (ValueError, TypeError):
            return default

    @staticmethod
    def _convert_baostock_data_to_records(symbol: str, bs_code: str, df: pd.DataFrame) -> List[dict]:
        """
        将BaoStock返回的DataFrame转换为stock_baostock_daily所需字段
        
        BaoStock返回的字段：
        date, code, open, high, low, close, preclose, volume, amount, adjustflag, turn, tradestatus, pctChg, isST
        
        :param symbol: 股票代码（原始格式，如"600000"）
        :param bs_code: BaoStock代码（如"sh.600000"）
        :param df: BaoStock返回的DataFrame
        :return: 转换后的记录列表
        """
        records: List[dict] = []
        if df is None or df.empty:
            return records
        
        for _, row in df.iterrows():
            try:
                # 解析日期
                trade_date_str = str(row.get('date', ''))
                if trade_date_str:
                    trade_date = datetime.strptime(trade_date_str, "%Y-%m-%d").date()
                else:
                    continue
                
                # 获取BaoStock原始数据
                record = {
                    "symbol": symbol,
                    "bs_code": bs_code,
                    "trade_date": trade_date,
                    "open": BaoStockDailyService._safe_float(row.get('open')),
                    "close": BaoStockDailyService._safe_float(row.get('close')),
                    "high": BaoStockDailyService._safe_float(row.get('high')),
                    "low": BaoStockDailyService._safe_float(row.get('low')),
                    "preclose": BaoStockDailyService._safe_float(row.get('preclose')),
                    "volume": BaoStockDailyService._safe_float(row.get('volume')),
                    "amount": BaoStockDailyService._safe_float(row.get('amount')),
                    "adjustflag": str(row.get('adjustflag', '')) if row.get('adjustflag') else None,
                    "turn": BaoStockDailyService._safe_float(row.get('turn')),
                    "tradestatus": str(row.get('tradestatus', '')) if row.get('tradestatus') else None,
                    "pct_chg": BaoStockDailyService._safe_float(row.get('pctChg')),
                    "is_st": str(row.get('isST', '')) if row.get('isST') else None,
                }
                records.append(record)
            except Exception as e:
                logger.warning(f"转换BaoStock数据失败，symbol={symbol}, date={row.get('date')}, error={e}")
                continue
        
        return records

    @classmethod
    async def fetch_daily_data_from_baostock(
        cls,
        symbol: str,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None
    ) -> List[dict]:
        """
        从BaoStock获取股票日K数据
        
        :param symbol: 股票代码，如 "600000"
        :param start_date: 开始日期，格式 "YYYY-MM-DD"，默认为None（获取全部）
        :param end_date: 结束日期，格式 "YYYY-MM-DD"，默认为None（获取全部）
        :return: 日K数据记录列表
        """
        # 在线程池中执行同步的BaoStock调用
        def _fetch_sync():
            # 登录BaoStock
            lg = bs.login()
            if lg.error_code != '0':
                raise ServiceException(f"BaoStock登录失败: {lg.error_msg}")
            
            try:
                # 转换股票代码格式
                bs_code = cls._convert_symbol_to_baostock_code(symbol)
                
                # 设置查询字段
                fields = "date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus,pctChg,isST"
                
                # 查询历史K线数据
                # adjustflag: 复权类型，1=后复权，2=前复权，3=不复权
                rs = bs.query_history_k_data_plus(
                    bs_code,
                    fields,
                    start_date=start_date or "",
                    end_date=end_date or "",
                    frequency="d",  # 日K线
                    adjustflag="3"  # 不复权，可根据需要修改
                )
                
                if rs.error_code != '0':
                    raise ServiceException(f"BaoStock查询失败: {rs.error_msg}")
                
                # 获取数据并转换为DataFrame
                data_list = []
                while (rs.error_code == '0') & rs.next():
                    data_list.append(rs.get_row_data())
                
                if not data_list:
                    return pd.DataFrame(), bs_code
                
                df = pd.DataFrame(data_list, columns=rs.fields)
                return df, bs_code
            finally:
                # 登出BaoStock
                bs.logout()
        
        # 在线程池中执行同步操作
        df, bs_code = await run_in_threadpool(_fetch_sync)
        
        # 转换数据格式
        records = cls._convert_baostock_data_to_records(symbol, bs_code, df)
        return records

    @classmethod
    async def _get_existing_dates(
        cls,
        query_db: AsyncSession,
        symbol: str,
        trade_dates: List[date]
    ) -> Set[date]:
        """
        获取已存在的交易日期集合
        
        :param query_db: 数据库会话
        :param symbol: 股票代码
        :param trade_dates: 交易日期列表
        :return: 已存在的交易日期集合
        """
        if not trade_dates:
            return set()
        
        existing_dates_list = await BaoStockDailyDao.get_existing_dates(query_db, symbol, trade_dates)
        return set(existing_dates_list)

    @classmethod
    async def sync_daily_data_to_db(
        cls,
        query_db: AsyncSession,
        symbol: str,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None
    ) -> CrudResponseModel:
        """
        从BaoStock获取数据并同步到数据库（自动过滤已存在的数据）
        
        :param query_db: 数据库会话
        :param symbol: 股票代码
        :param start_date: 开始日期，格式 "YYYY-MM-DD"
        :param end_date: 结束日期，格式 "YYYY-MM-DD"
        :return: 同步结果
        """
        try:
            # 从BaoStock获取数据
            records = await cls.fetch_daily_data_from_baostock(symbol, start_date, end_date)
            
            if not records:
                return CrudResponseModel(is_success=True, message=f'股票 {symbol} 没有获取到数据')
            
            # 提取交易日期
            trade_dates = [record['trade_date'] for record in records if record.get('trade_date')]
            
            # 检查哪些数据已存在
            existing_dates = await cls._get_existing_dates(query_db, symbol, trade_dates)
            
            # 过滤出需要插入的新数据
            new_records = [
                record for record in records
                if record.get('trade_date') and record['trade_date'] not in existing_dates
            ]
            
            if not new_records:
                return CrudResponseModel(
                    is_success=True,
                    message=f'股票 {symbol} 数据已存在，无需同步'
                )
            
            # 批量插入新数据到BaoStock表
            await BaoStockDailyDao.batch_add_baostock_daily_dao(query_db, new_records)
            await query_db.commit()
            
            return CrudResponseModel(
                is_success=True,
                message=f'成功同步 {len(new_records)} 条新数据（共获取 {len(records)} 条，{len(existing_dates)} 条已存在）'
            )
        except Exception as e:
            await query_db.rollback()
            logger.error(f"同步BaoStock数据失败: {e}")
            raise ServiceException(f"同步BaoStock数据失败: {str(e)}")

    @classmethod
    async def get_daily_data_from_baostock(
        cls,
        symbol: str,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None
    ) -> List[dict]:
        """
        从BaoStock获取股票日K数据并返回（不保存到数据库）
        
        :param symbol: 股票代码
        :param start_date: 开始日期，格式 "YYYY-MM-DD"
        :param end_date: 结束日期，格式 "YYYY-MM-DD"
        :return: 日K数据列表
        """
        records = await cls.fetch_daily_data_from_baostock(symbol, start_date, end_date)
        
        # 转换为前端需要的格式
        result = []
        for record in records:
            result.append({
                'symbol': record['symbol'],
                'bsCode': record.get('bs_code'),
                'tradeDate': record['trade_date'].strftime('%Y-%m-%d') if record['trade_date'] else None,
                'open': float(record['open']) if record['open'] is not None else None,
                'close': float(record['close']) if record['close'] is not None else None,
                'high': float(record['high']) if record['high'] is not None else None,
                'low': float(record['low']) if record['low'] is not None else None,
                'preclose': float(record['preclose']) if record.get('preclose') is not None else None,
                'volume': float(record['volume']) if record.get('volume') is not None else None,
                'amount': float(record['amount']) if record['amount'] is not None else None,
                'adjustflag': record.get('adjustflag'),
                'turn': float(record['turn']) if record['turn'] is not None else None,
                'tradestatus': record.get('tradestatus'),
                'pctChg': float(record['pct_chg']) if record['pct_chg'] is not None else None,
                'isSt': record.get('is_st'),
            })
        
        return result

    @classmethod
    async def get_baostock_daily_page(
        cls,
        query_db: AsyncSession,
        symbol: str,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        page_num: int = 1,
        page_size: int = 10
    ):        
        """
        分页查询BaoStock股票日K数据
        
        :param query_db: 数据库会话
        :param symbol: 股票代码
        :param start_date: 开始日期，格式 "YYYY-MM-DD"
        :param end_date: 结束日期，格式 "YYYY-MM-DD"
        :param page_num: 当前页码
        :param page_size: 每页记录数
        :return: 分页结果
        """
        # 构建查询条件
        from sqlalchemy import select
        from sqlalchemy.sql import and_
        from module_stock.entity.do.baostock_daily_do import StockBaoStockDaily
        
        conditions = [StockBaoStockDaily.symbol == symbol]
        
        # 添加日期条件
        if start_date:
            conditions.append(StockBaoStockDaily.trade_date >= datetime.strptime(start_date, "%Y-%m-%d").date())
        if end_date:
            conditions.append(StockBaoStockDaily.trade_date <= datetime.strptime(end_date, "%Y-%m-%d").date())
        
        # 构建查询
        query = select(StockBaoStockDaily).where(and_(*conditions))
        query = query.order_by(StockBaoStockDaily.trade_date.desc())
        
        # 分页处理
        from utils.page_util import PageUtil
        result = await PageUtil.paginate(
            db=query_db,
            query=query,
            page_num=page_num,
            page_size=page_size,
            is_page=True
        )
        
        # 转换为前端需要的格式
        items = []
        # 处理分页结果，注意PageUtil.paginate返回的是PageResponseModel对象，其数据在rows属性中
        data_rows = result.rows if hasattr(result, 'rows') else result
        for item in data_rows:
            # 处理item可能是字典或对象的情况
            if isinstance(item, dict):
                # 字典形式
                items.append({
                    'id': item.get('id'),
                    'symbol': item.get('symbol'),
                    'bsCode': item.get('bsCode'),
                    'tradeDate': item.get('tradeDate'),
                    'open': item.get('open'),
                    'close': item.get('close'),
                    'high': item.get('high'),
                    'low': item.get('low'),
                    'preclose': item.get('preclose'),
                    'volume': item.get('volume'),
                    'amount': item.get('amount'),
                    'adjustflag': item.get('adjustflag'),
                    'turn': item.get('turn'),
                    'tradestatus': item.get('tradestatus'),
                    'pctChg': item.get('pctChg'),
                    'isSt': item.get('isSt'),
                    'createTime': item.get('createTime'),
                    'updateTime': item.get('updateTime')
                })
            else:
                # 对象形式
                items.append({
                    'id': item.id,
                    'symbol': item.symbol,
                    'bsCode': item.bs_code,
                    'tradeDate': item.trade_date.strftime('%Y-%m-%d') if item.trade_date else None,
                    'open': float(item.open) if item.open is not None else None,
                    'close': float(item.close) if item.close is not None else None,
                    'high': float(item.high) if item.high is not None else None,
                    'low': float(item.low) if item.low is not None else None,
                    'preclose': float(item.preclose) if item.preclose is not None else None,
                    'volume': float(item.volume) if item.volume is not None else None,
                    'amount': float(item.amount) if item.amount is not None else None,
                    'adjustflag': item.adjustflag,
                    'turn': float(item.turn) if item.turn is not None else None,
                    'tradestatus': item.tradestatus,
                    'pctChg': float(item.pct_chg) if item.pct_chg is not None else None,
                    'isSt': item.is_st,
                    'createTime': item.create_time,
                    'updateTime': item.update_time
                })
        
        # 更新分页结果中的items
        if hasattr(result, 'rows'):
            result.rows = items
        else:
            result = items
        return result

    @classmethod
    async def get_baostock_daily_detail(
        cls,
        query_db: AsyncSession,
        id: int
    ) -> dict:
        """
        获取BaoStock股票日K数据详情
        
        :param query_db: 数据库会话
        :param id: 数据ID
        :return: 数据详情
        """
        daily_info = await BaoStockDailyDao.get_baostock_daily_detail_by_id(query_db, id)
        if not daily_info:
            raise ServiceException(f"ID为{id}的数据不存在")
        
        # 转换为前端需要的格式
        return {
            'id': daily_info.id,
            'symbol': daily_info.symbol,
            'bsCode': daily_info.bs_code,
            'tradeDate': daily_info.trade_date.strftime('%Y-%m-%d') if daily_info.trade_date else None,
            'open': float(daily_info.open) if daily_info.open is not None else None,
            'close': float(daily_info.close) if daily_info.close is not None else None,
            'high': float(daily_info.high) if daily_info.high is not None else None,
            'low': float(daily_info.low) if daily_info.low is not None else None,
            'preclose': float(daily_info.preclose) if daily_info.preclose is not None else None,
            'volume': float(daily_info.volume) if daily_info.volume is not None else None,
            'amount': float(daily_info.amount) if daily_info.amount is not None else None,
            'adjustflag': daily_info.adjustflag,
            'turn': float(daily_info.turn) if daily_info.turn is not None else None,
            'tradestatus': daily_info.tradestatus,
            'pctChg': float(daily_info.pct_chg) if daily_info.pct_chg is not None else None,
            'isSt': daily_info.is_st,
            'createTime': daily_info.create_time,
            'updateTime': daily_info.update_time
        }

    @classmethod
    async def delete_baostock_daily(
        cls,
        query_db: AsyncSession,
        ids: str
    ) -> CrudResponseModel:
        """
        删除BaoStock股票日K数据
        
        :param query_db: 数据库会话
        :param ids: 需要删除的ID，多个用逗号分隔
        :return: 删除结果
        """
        try:
            # 解析ID列表
            id_list = [int(id.strip()) for id in ids.split(',') if id.strip().isdigit()]
            if not id_list:
                raise ServiceException("请提供有效的ID")
            
            # 批量删除
            from sqlalchemy import delete
            from module_stock.entity.do.baostock_daily_do import StockBaoStockDaily
            
            stmt = delete(StockBaoStockDaily).where(StockBaoStockDaily.id.in_(id_list))
            result = await query_db.execute(stmt)
            await query_db.commit()
            
            return CrudResponseModel(
                is_success=True,
                message=f"成功删除 {result.rowcount} 条数据"
            )
        except Exception as e:
            await query_db.rollback()
            logger.error(f"删除BaoStock数据失败: {e}")
            raise ServiceException(f"删除失败: {str(e)}")

    @classmethod
    async def batch_sync_daily_data_to_db(
        cls,
        query_db: AsyncSession,
        symbols: str,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None
    ) -> List[dict]:
        """
        批量从BaoStock获取股票日K数据并同步到数据库
        
        :param query_db: 数据库会话
        :param symbols: 股票代码列表，多个用逗号分隔
        :param start_date: 开始日期，格式 "YYYY-MM-DD"
        :param end_date: 结束日期，格式 "YYYY-MM-DD"
        :return: 批量同步结果列表
        """
        # 解析股票代码列表
        symbol_list = [symbol.strip() for symbol in symbols.split(',') if symbol.strip()]
        if not symbol_list:
            raise ServiceException("请提供有效的股票代码")
        
        # 逐个同步每个股票的数据
        results = []
        for symbol in symbol_list:
            try:
                # 同步单个股票数据
                result = await cls.sync_daily_data_to_db(
                    query_db=query_db,
                    symbol=symbol,
                    start_date=start_date,
                    end_date=end_date
                )
                results.append({
                    'symbol': symbol,
                    'success': True,
                    'message': result.message
                })
            except Exception as e:
                logger.error(f"批量同步股票 {symbol} 数据失败: {e}")
                results.append({
                    'symbol': symbol,
                    'success': False,
                    'message': str(e)
                })
        
        return results

    @classmethod
    async def get_all_symbols(
        cls,
        query_db: AsyncSession
    ) -> List[str]:
        """
        获取已保存的BaoStock日K数据的股票代码列表
        
        :param query_db: 数据库会话
        :return: 股票代码列表
        """
        return await BaoStockDailyDao.get_all_symbols(query_db)

    @classmethod
    async def _safe_fetch_with_retry(
        cls,
        symbol: str,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        max_retries: int = 3
    ) -> List[dict]:
        """
        带重试机制的安全获取BaoStock数据方法
        
        :param symbol: 股票代码
        :param start_date: 开始日期，格式 "YYYY-MM-DD"
        :param end_date: 结束日期，格式 "YYYY-MM-DD"
        :param max_retries: 最大重试次数
        :return: 日K数据列表
        """
        import asyncio
        retries = 0
        while retries < max_retries:
            try:
                return await cls.fetch_daily_data_from_baostock(symbol, start_date, end_date)
            except Exception as e:
                retries += 1
                if retries >= max_retries:
                    logger.error(f"获取BaoStock数据失败，已重试{max_retries}次: {e}")
                    raise
                
                # 指数退避重试
                wait_time = (2 ** retries) * 0.5
                logger.warning(f"获取BaoStock数据失败，{retries}秒后重试 ({retries}/{max_retries}): {e}")
                await asyncio.sleep(wait_time)

